package cn.bud.msgbus.consumer.server;

import cn.bud.msgbus.server.bean.MsgBody;
import cn.bud.msgbus.server.bean.Subscriber;
import cn.bud.msgbus.utils.BusUtils;
import cn.hutool.core.collection.ConcurrentHashSet;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MsgBusServer
 *
 * @author hxj
 * @version 1.0
 * @description
 * @date 2024-06-19 11:05
 */
@Slf4j
public class MsgBusServer {

    private static final String LOG_TAG = "消息总线服务端";

    private int serverMode = 0;
    private int[] nettyPorts;
    MsgBusDataQueue msgBusDataQueue;


    // subscriber地址(唯一标识) ==> 订阅者实例对象
    private static Map<String, Subscriber> subscriberMap = new ConcurrentHashMap<>();

    // 主题 ==> 订阅者地址(唯一标识)
    private static Map<String, Set<String>> topicSubMap = new ConcurrentHashMap<>();

    private MsgBusServer() {}

    public void addSubscriber(Subscriber newSub) {
        Subscriber oldSub = subscriberMap.get(newSub.name());
        if (oldSub == null) {
            subscriberMap.put(newSub.name(), newSub);
            for (String topic : newSub.topicSet()) {
                addTopic(topic, newSub.name());
            }
            return;
        }
        List<String> delTopics = new ArrayList<>();
        List<String> addTopics = new ArrayList<>();
        for (String oldTopic : oldSub.topicSet()) {
            if (!newSub.topicSet().contains(oldTopic)) {
                delTopics.add(oldTopic);
            }
        }
        for (String newTopic : newSub.topicSet()) {
            if (oldSub.topicSet().contains(newTopic)) {
                addTopics.add(newTopic);
            }
        }
        if (CollectionUtils.isNotEmpty(delTopics)) {
            for (String delTopic : delTopics) {
                delSubscriber(delTopic, newSub.name());
            }
        }
        if (CollectionUtils.isNotEmpty(addTopics)) {
            for (String addTopic : addTopics) {
                addTopic(addTopic, newSub.name());
            }
        }
    }

    private void addTopic(String topic, String subName) {
        Set<String> nameSet = topicSubMap.get(topic);
        if (nameSet == null) {
            nameSet = new ConcurrentHashSet<>();
            topicSubMap.put(topic, nameSet);
        }
        nameSet.add(subName);
    }

    public void delSubscriber(String topic, String subName) {
        Set<String> nameSet = topicSubMap.get(topic);
        if (nameSet == null) {
           return;
        }
        nameSet.remove(subName);
        if (nameSet.isEmpty()) {
            synchronized (topicSubMap) {
                topicSubMap.remove(topic);
            }
            synchronized (subscriberMap) {
                subscriberMap.remove(subName);
            }
        }
    }
    public void delSubscriber(String subName) {
        Subscriber subscriber = subscriberMap.get(subName);
        if (CollectionUtils.isEmpty(subscriber.topicSet())) {
            return;
        }
        for (String topic : subscriber.topicSet()) {
            delSubscriber(topic, subName);
        }
    }

    public List<Subscriber> listSubByTopic(String topic) {
        List<Subscriber> list = new ArrayList<>();
        Set<String> subNameSet = topicSubMap.get(topic);
        if (CollectionUtils.isNotEmpty(subNameSet)) {
            for (String subName : subNameSet) {
                list.add(subscriberMap.get(subName));
            }
        }
        return list;
    }


    Thread workThread;
    boolean stop = false;

    public void start() {

        // 启用本地消息推送管理线程
        workThread = new Thread(()-> {
            while (!stop) {
                try {
                    sendData();
                }catch (Exception e) {
                    if (e.getMessage() == null) {
                        log.warn("{} 发生异常：{}", LOG_TAG, e);
                    }else {
                        log.warn("{} 发生异常：{}", LOG_TAG, e.getMessage());

                    }
                }
            }
        });
        workThread.start();

        // TODO 启动netty服务

    }


    public void stop() {
        stop = true;
        msgBusDataQueue.stop();
    }


    private void sendData() {
        MsgBody msg = msgBusDataQueue.poll();
        if (msg == null) {
            BusUtils.sleep(5);
            return;
        }
        List<Subscriber> subscribers = listSubByTopic(msg.getTopic());
        for (Subscriber subscriber : subscribers) {
            try {
                subscriber.sendMsg(msg);
            }catch (Exception e) {
                log.warn("{} {}推送异常, 数据：{}， 异常描述：{}",
                        LOG_TAG, subscriber.name(), msg, e.getMessage());
            }
        }
    }

    public static class MsgBusServerBuilder {
        MsgBusServer server = new MsgBusServer();

        public MsgBusServerBuilder serverMode(Integer serverMode) {
            server.serverMode = serverMode;
            return this;
        }
        public MsgBusServerBuilder nettyPorts(int[] nettyPorts) {
            server.nettyPorts = nettyPorts;
            return this;
        }
        public MsgBusServerBuilder msgBusDataQueue(MsgBusDataQueue msgBusDataQueue) {
            server.msgBusDataQueue = msgBusDataQueue;
            return this;
        }
        public <T extends Subscriber> MsgBusServerBuilder subscriber(T... subscriber) {
            if (ArrayUtils.isNotEmpty(subscriber)) {
                for (T subscriber1 : subscriber) {
                    server.addSubscriber(subscriber1);
                }
            }
            return this;
        }
        public <T extends Subscriber> MsgBusServerBuilder subscribers(List<T> subscribers) {
            if (CollectionUtils.isNotEmpty(subscribers)) {
                for (Subscriber subscriber : subscribers) {
                    server.addSubscriber(subscriber);
                }
            }
            return this;
        }

        public MsgBusServer build() {
            return server;
        }
    }
}
